package com.desheng.bigdata.flink.stream;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.io.Serializable;

/**
 * 流式的编程
 *  入口：StreamExecutionEnvironment
 */
public class JavaFlinkStreamWordCountApp {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment sEnv = StreamExecutionEnvironment.getExecutionEnvironment();

        //加载数据
        DataStreamSource<String> inputs = sEnv.socketTextStream("bigdata01", 9999);

        SingleOutputStreamOperator<Tuple2<String, Integer>> pairs = inputs.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                String[] words = line.split("\\s+");
                for (String word : words) {
                    collector.collect(new Tuple2<String, Integer>(word, 1));
                }
            }
        });

        pairs.keyBy(0)
                .sum(1)
                .print();


        //
        sEnv.execute(JavaFlinkStreamWordCountApp.class.getSimpleName());
    }
}

class JavaWordCount implements Serializable{
    private String word;
    private int count;

    public JavaWordCount(String word, int count) {
        this.word = word;
        this.count = count;
    }

    public JavaWordCount() {
    }

    public String getWord() {
        return word;
    }

    public void setWord(String word) {
        this.word = word;
    }

    public int getCount() {
        return count;
    }

    public void setCount(int count) {
        this.count = count;
    }

    @Override
    public String toString() {
        return "JavaWordCount{" +
                "word='" + word + '\'' +
                ", count=" + count +
                '}';
    }
}
